package com.boat.starter.pulsar.config;

import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

import org.apache.pulsar.client.api.PulsarClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.StringUtils;

import com.boat.starter.pulsar.annotation.PulsarMessageListener;
import com.boat.starter.pulsar.common.PulsarMessageListenerInfo;
import com.boat.starter.pulsar.constant.MQConstant;
import com.boat.starter.pulsar.core.PulsarConsumerFactory;
import com.boat.starter.pulsar.core.PulsarListener;
import com.boat.starter.pulsar.properties.MQProperties;

import cn.hutool.core.util.StrUtil;

/**
 * Description: 消费监听容器配置
 *
 * @author: zuomin (myleszelic@outlook.com)
 * @date: 2021/07/20-10:39
 */
@Configuration
@ConditionalOnProperty(prefix = MQConstant.PREFIX, value = {"pulsar.serviceUrl", "pulsar.authenticationToken"})
public class ListenerContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {

    private static Logger log = LoggerFactory.getLogger(ListenerContainerConfiguration.class);

    public static final String PULSAR_CLIENT_BEAN_NAME = "defaultPulsarClient";


    private ConfigurableApplicationContext applicationContext;

    private final StandardEnvironment environment;

    private final MQProperties mqProperties;

    private final AtomicLong counter = new AtomicLong(0);

    private final PulsarClient client;

    private MessageConverter messageConverter;

    public ListenerContainerConfiguration(StandardEnvironment environment, MQProperties mqProperties, @Qualifier(value = PULSAR_CLIENT_BEAN_NAME) PulsarClient client) {
        this.environment = environment;
        this.mqProperties = mqProperties;
        this.client = client;
    }

    @Override
    public void afterSingletonsInstantiated() {
        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(PulsarMessageListener.class)
                .entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        beans.forEach(this::registerContainer);
    }

    private void registerContainer(String beanName, Object bean) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
        if (!PulsarListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " cannot be instance of " + PulsarListener.class.getName());
        }
        PulsarMessageListener annotation = clazz.getAnnotation(PulsarMessageListener.class);
        PulsarMessageListenerInfo annotationInfo = new PulsarMessageListenerInfo(annotation);
        //支持占位符，从配置读取值
        setDynamicValue(annotationInfo);

        validate(annotationInfo);

        String containerBeanName = String.format("%s_%s", DefaultPulsarListenerContainer.class.getName(), counter.incrementAndGet());
        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;

        genericApplicationContext.registerBean(containerBeanName, DefaultPulsarListenerContainer.class,
                () -> createPulsarListenerContainer(containerBeanName, bean, annotationInfo));

        DefaultPulsarListenerContainer container = genericApplicationContext.getBean(containerBeanName,
                DefaultPulsarListenerContainer.class);
        if (!container.isRunning()) {
            try {
                container.start();
            } catch (Exception e) {
                log.error("Started container failed. {}", container, e);
                throw new RuntimeException(e);
            }
        }
        log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
    }

    /**
     * 支持动态设置注解值
     *
     * @param annotationInfo
     */
    private void setDynamicValue(PulsarMessageListenerInfo annotationInfo) {
    	if(annotationInfo.getTopic().contains("$")) {
    		String fixTopic = this.applicationContext.getEnvironment().resolvePlaceholders(annotationInfo.getTopic());
    		if (fixTopic.contains("$")) {
				// 此时说明配置中没有此配置项，但是代码中写了需要用到此配置。重置为空
    			fixTopic = "";
			}
			annotationInfo.setTopic(fixTopic);
    	}
		
		if(StrUtil.isBlank(annotationInfo.getSubscriptionName())) {
			String appName = this.applicationContext.getEnvironment().getProperty("spring.application.name");
			annotationInfo.setSubscriptionName(appName);
		}
        if(annotationInfo.getSubscriptionName().contains("$")) {
        	String fixSubscriptionName = this.applicationContext.getEnvironment().resolvePlaceholders(annotationInfo.getSubscriptionName());
        	if (fixSubscriptionName.contains("$")) {
				// 此时说明配置中没有此配置项，但是代码中写了需要用到此配置。重置为空
        		fixSubscriptionName = "";
			}
        	annotationInfo.setSubscriptionName(fixSubscriptionName);
    	}
		
        if(annotationInfo.getTags().contains("$")) {
        	String fixTags = this.applicationContext.getEnvironment().resolvePlaceholders(annotationInfo.getTags());
			if (fixTags.contains("$")) {
				// 此时说明配置中没有此配置项，但是代码中写了需要用到此配置。重置为空
				fixTags = "";
			}
			annotationInfo.setTags(fixTags);
    	}
    }

    private DefaultPulsarListenerContainer createPulsarListenerContainer(String containerBeanName, Object bean, PulsarMessageListenerInfo annotationInfo) {

        DefaultPulsarListenerContainer listenerContainer = new DefaultPulsarListenerContainer();
        listenerContainer.setName(containerBeanName);
        listenerContainer.setClient(client);
        listenerContainer.setMessageListenerInfo(annotationInfo);
        listenerContainer.setPulsarListener((PulsarListener) bean);
        listenerContainer.setConsumerFactory(new PulsarConsumerFactory(client, environment, mqProperties.getPulsar()));
        return listenerContainer;
    }

    private void validate(PulsarMessageListenerInfo annotationInfo) {
        if (!StringUtils.hasLength(annotationInfo.getTopic()) || !StringUtils.hasLength(annotationInfo.getSubscriptionName())) {
            throw new BeanDefinitionValidationException(
                    "Bad annotation definition in @PulsarMessageListener, topic or subscriptionName cannot be null");
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }

}
